package cn.com.yang.es;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.config.RequestConfig.Builder;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;



/**
 * ES创建索引
 * TODO
 * @author yangyijin
 * @since 2024年1月6日-下午5:01:17
 */
@Component
public class EsCreateIndexDemo {
	@Autowired
	RestHighLevelClient restHighLevelClient;
	public static void main(String[] args) {
		try {
			createIndex();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	private static void createIndex() throws IOException {
		HttpHost host = new HttpHost("localhost", 9400);
		RestClientBuilder builder = RestClient.builder(host);
		//创建请求
		RestHighLevelClient client = new RestHighLevelClient(builder);
		//定义索引
		String index = "index_name";
		String type ="typeName";
		
//		XContentBuilder  contentBuilder = XContentFactory.jsonBuilder();
//		contentBuilder.startObject();
//		{
//			contentBuilder.startObject("properties");
//			{
//				contentBuilder.startObject("fieldName");
//				{
//					contentBuilder.field("type","text");//字段类型
//				}
//				contentBuilder.endObject();
//			}
//			contentBuilder.endObject();
//		}
//		contentBuilder.endObject();
		
		//创建索引请求
		CreateIndexRequest request = new CreateIndexRequest(index);
//      可以使用CreateIndexRequest或PutIndexTemplateRequest来创建索引。
		
//		PutIndexTemplateRequest indexTemplateRequest = new PutIndexTemplateRequest("templateName");
//		indexTemplateRequest.settings(Settings.builder().put("index.number_of_shards", 1) // 分片数
//		        .put("index.number_of_replicas", 0)); // 副本数
//		indexTemplateRequest.mapping(type, XContentType.JSON);
//		indexTemplateRequest.mapping(contentBuilder);
//		String  jsonStr ="";
//		request.source(jsonStr,XContentType.JSON);
		request.settings(Settings.builder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0));
//		request.mapping(type,contentBuilder);
		CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
		
		// 异步调用，需要设置监听器
		client.indices().createAsync(request, RequestOptions.DEFAULT, null);//listener
		
		//1.添加数据
		IndexRequest idxRequest = new IndexRequest(index).id("id");//id是逐渐
		idxRequest.source(new HashMap(),XContentType.JSON);//添加数据
		IndexResponse response= client.index(idxRequest, RequestOptions.DEFAULT);
		//异步添加
		client.indexAsync(idxRequest, RequestOptions.DEFAULT, null);//listener
		
		
		//2.更新数据
		UpdateRequest updRequest = new UpdateRequest(index, type);
		updRequest.doc(new HashMap(),XContentType.JSON);//更新数据
		// 进行同步请求更新
		UpdateResponse updResponse = client.update(updRequest, RequestOptions.DEFAULT);

		// 异步请求更新，需要设置监听器
		client.updateAsync(updRequest, RequestOptions.DEFAULT, null);//listener
		
		//3.删除数据
		DeleteRequest delRequest = new DeleteRequest(index);
		delRequest.id("id");//删除的id
		// 同步删除数据
		DeleteResponse delresponse = client.delete(delRequest, RequestOptions.DEFAULT);

		// 异步删除数据，需要设置监听器
		client.deleteAsync(delRequest, RequestOptions.DEFAULT, null);//listener
		
		//4.批量操作
		BulkRequest bulkRequest = new BulkRequest();
		bulkRequest.add(delRequest);//可添加多个
		bulkRequest.add(updRequest);//可添加多个
		bulkRequest.add(idxRequest);//可添加多个
		client.bulk(bulkRequest, RequestOptions.DEFAULT);
		client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, null);//listener
	}
	/**
	 * 
	 * @return
	 */
	public RestHighLevelClient instance() {
		List<HttpHost> hostList = new ArrayList<>();
		String address = "127.0.0.1:9400,192.168.1.2:9400";
		StringTokenizer tokenizer = new StringTokenizer(address,",");
		while(tokenizer.hasMoreTokens()) {
			String addr = tokenizer.nextToken();
			String[] arr = addr.split(":");
			hostList.add(new HttpHost(arr[0],Integer.valueOf(arr[1]),"http"));
		}
		RestClientBuilder restClientBuilder = RestClient.builder(hostList.<HttpHost>toArray(new HttpHost[0]));
		//设置请求超时
		restClientBuilder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
			@Override
			public Builder customizeRequestConfig(Builder requestConfigBuilder) {
				requestConfigBuilder.setConnectTimeout(6000);
				requestConfigBuilder.setSocketTimeout(10000);
				requestConfigBuilder.setConnectionRequestTimeout(6000);
				return requestConfigBuilder;
			}
		});
		//设置服务密码
		restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
			
			@Override
			public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
				httpClientBuilder.setMaxConnTotal(100);//最大连接数
				httpClientBuilder.setMaxConnPerRoute(100);//最大单个路由连接数
				//设置用户名密码
				BasicCredentialsProvider basicCredentialsProvider = new BasicCredentialsProvider();
				basicCredentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("elastic", "Aa123456"));
				httpClientBuilder.setDefaultCredentialsProvider(basicCredentialsProvider);
				return httpClientBuilder;
			}
		});
		return new RestHighLevelClient(restClientBuilder);
	}	
	
	/*
	 * 创建ES索引
	 * @param indexName 索引名称
	 * @param jsonMapping 索引对象文件
	 * @return
	 */
	@SuppressWarnings("resource")
	public boolean createIndex(String indexName,String jsonMapping) {
		//构建ES客户端
		HttpHost host = new HttpHost("localhost", 9400);
		RestClientBuilder builder = RestClient.builder(host);
		RestHighLevelClient client = new RestHighLevelClient(builder);
		
		//构建创建索引对象
		CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
		createIndexRequest.source(jsonMapping,XContentType.JSON);
		try {
			CreateIndexResponse indexResponse = 
					client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
			return indexResponse.isAcknowledged();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return false;
	}
	/**
	 * 批量新增
	 * @param indexName 索引名称
	 * @param listData 数据
	 * @param idFiled  主键id字段
	 * @return
	 */
	public boolean batchAdd(String indexName, List<Map<String,Object>> listData,String idFiled) {
		BulkRequest bulkRequest = new BulkRequest();
		for(Map<String,Object> single: listData) {
			String id = (String)single.get(idFiled);
			IndexRequest indexRequest = new IndexRequest(indexName).id(id);
			indexRequest.source(single,XContentType.JSON);
			bulkRequest.add(indexRequest);
		}
		try {
			BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
			if(response.hasFailures()) {
				System.out.print(response.buildFailureMessage());
				return false;
			}
			return true;
		} catch (IOException e) {
			e.printStackTrace();
		}
		return false;
	}
	
	
	/**
	 * 批量更新
	 * @param indexName 索引名称
	 * @param listData 数据
	 * @param idFiled 主键
	 * @return
	 */
	public boolean batchUpdate(String indexName, List<Map<String,Object>> listData,String idFiled) {
		BulkRequest bulkRequest = new BulkRequest();
		for(Map<String,Object> single: listData) {
			String id = (String)single.get(idFiled);
			UpdateRequest updateRequest = new UpdateRequest(indexName,id);
			updateRequest.doc(single,XContentType.JSON);
			bulkRequest.add(updateRequest);
		}
		try {
			BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
			if(response.hasFailures()) {
				System.out.print(response.buildFailureMessage());
				return false;
			}
			return true;
		} catch (IOException e) {
			e.printStackTrace();
		}
		return false;
	}
	
	/**
	 * 批量删除
	 * @param indexName 索引名称
	 * @param ids 数据主键集合
	 * @return
	 */
	public boolean batchDelete(String indexName,List<String> ids) {
		BulkRequest bulkRequest = new BulkRequest();
		for(String id: ids) {
			DeleteRequest deleteRequest = new DeleteRequest(indexName, id);
			bulkRequest.add(deleteRequest);
		}
		try {
			BulkResponse response = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
			if(response.hasFailures()) {
				System.out.print(response.buildFailureMessage());
				return false;
			}
			return true;
		} catch (IOException e) {
			e.printStackTrace();
		}
		return false;
	}
}
